/*
 ***************************************************************************************
 * All rights Reserved, Designed By RTF
 * @Title:  RocketMqManager.java
 * @Package com.rtf.repair.mq.rocketmq.producer
 * @Description: rocketmq 生产者管理类，管理其生命周期
 * @author: sfl
 * @date:   2020-10-17 16:39:00
 * @version V1.0
 * @Copyright: 2020 RTF. All rights reserved.
 * 注意：本内容仅限于公司内部使用，禁止外泄以及用于其他的商业目
 *  ----------------------------------------------------------------------------------
 * 文件修改记录
 *     文件版本：         修改人：             修改原因：
 ***************************************************************************************
 */
package com.rtf.framework.mq.rocketmq.producer;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.springframework.util.Assert;

import com.rtf.framework.mq.common.MQException;
import com.rtf.framework.mq.common.Service;

import lombok.extern.slf4j.Slf4j;

/**
 *
 * @Description: rocketmq 生产者管理类，管理其生命周期
 * @ClassName:  RocketMqManager
 * @author: sfl
 * @date:   2020-10-20 11:38:25
 * @since:  v1.0
 */
@Slf4j
public class RocketMqManager implements Service {

	private RocketMqProducerProperties properties;

	private DefaultMQProducer producer;

	/**
	 * Instantiates a new rocket mq manager.
	 *
	 * @param properties the properties
	 */
	public RocketMqManager(RocketMqProducerProperties properties) {
		Assert.notNull(properties, "RocketMqProducerProperties is null ");
		this.properties = properties;
	}

	@Override
	public void init() {

		Assert.isTrue(StringUtils.isNotBlank(properties.getNamesrvAddr()), "Rocketmq producer namesrvAddr is null");
		Assert.isTrue(StringUtils.isNotBlank(properties.getInstanceName()), "Rocketmq producer intanceName is null");
		this.producer = new DefaultMQProducer();
		this.producer.setNamesrvAddr(properties.getNamesrvAddr());
		this.producer.setInstanceName(properties.getInstanceName());
		this.producer.setPollNameServerInterval(properties.getPollNameServerInterval());
		this.producer.setHeartbeatBrokerInterval(properties.getHeartbeatBrokerInterval());
		this.producer.setSendMsgTimeout(properties.getSendMsgTimeout());
		this.producer.setMaxMessageSize(properties.getMaxMessageSize());
		this.producer.setRetryTimesWhenSendFailed(properties.getRetryTimesWhenSendFailed());
		this.producer.setRetryTimesWhenSendAsyncFailed(properties.getRetryTimesWhenSendFailed());
		this.producer.setVipChannelEnabled(false);
		if (StringUtils.isNotBlank(properties.getGroupName()))
			this.producer.setProducerGroup(properties.getGroupName());
	}

	@Override
	public void start() {
		try {
			producer.start();
		} catch (MQClientException e) {
			throw new MQException("Rocketmq producer starting failed", e);
		}

	}

	@Override
	public void destroy() {
		producer.shutdown();

	}

	public MQProducer getProducer() {
		return producer;
	}

}
